﻿//******************************************************************************************************
//  IaonSession.cs - Gbtc
//
//  Copyright © 2010, Grid Protection Alliance.  All Rights Reserved.
//
//  Licensed to the Grid Protection Alliance (GPA) under one or more contributor license agreements. See
//  the NOTICE file distributed with this work for additional information regarding copyright ownership.
//  The GPA licenses this file to you under the Eclipse Public License -v 1.0 (the "License"); you may
//  not use this file except in compliance with the License. You may obtain a copy of the License at:
//
//      http://www.opensource.org/licenses/eclipse-1.0.php
//
//  Unless agreed to in writing, the subject software distributed under the License is distributed on an
//  "AS-IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. Refer to the
//  License for the specific language governing permissions and limitations.
//
//  Code Modification History:
//  ----------------------------------------------------------------------------------------------------
//  2/2013 - University of Illinois, Information Trust Institute, Coordinated Science Laboratory
//       Updated code to work for the SIEGate system and architecture
//
//  08/23/2011 - J. Ritchie Carroll
//       Generated original version of source code.
//
//******************************************************************************************************

using GSF.Collections;
using GSF.Configuration;
using System;
using System.Collections;
using System.Collections.Concurrent;
using System.Data;
using System.Linq;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
using SIEGate.Core.Legacy;
using SIEGate.Core;
using System.Reflection;
using GSF.IO;
using System.IO;
using SIEGate.Core.Configurator;
using System.Collections.Generic;
using SIEGate.Core.MessageBusses;
using SIEGate.Core.Notifications;

namespace GSF.TimeSeries.Adapters
{
    /// <summary>
    /// Represents a new Input, Action, Output interface session.
    /// </summary>
    public class IaonSession : IProvideStatus, IDisposable
    {
        #region [ Members ]

        // Events

        /// <summary>
        /// Provides status messages to consumer.
        /// </summary>
        /// <remarks>
        /// <see cref="EventArgs{T1,T2}.Argument1"/> is the new status message.<br/>
        /// <see cref="EventArgs{T1,T2}.Argument2"/> is the message <see cref="UpdateType"/>.
        /// </remarks>
        public event EventHandler<EventArgs<string, UpdateType>> StatusMessage;

        /// <summary>
        /// Event is raised when there is an exception encountered while processing.
        /// </summary>
        /// <remarks>
        /// <see cref="EventArgs{T}.Argument"/> is the exception that was thrown.
        /// </remarks>
        public event EventHandler<EventArgs<Exception>> ProcessException;

        /// <summary>
        /// Event is raised when <see cref="ILegacyAdapter.InputMeasurementKeys"/> are updated.
        /// </summary>
        public event EventHandler InputMeasurementKeysUpdated;

        /// <summary>
        /// Event is raised when <see cref="ILegacyAdapter.OutputMeasurements"/> are updated.
        /// </summary>
        public event EventHandler OutputMeasurementsUpdated;

        /// <summary>
        /// Event is raised every five seconds allowing consumer to track current number of unpublished seconds of data in the queue.
        /// </summary>
        /// <remarks>
        /// <see cref="EventArgs{T}.Argument"/> is the total number of unpublished seconds of data.
        /// </remarks>
        public event EventHandler<EventArgs<int>> UnpublishedSamples;

        /// <summary>
        /// Event is raised every five seconds allowing host to track total number of unprocessed measurements.
        /// </summary>
        /// <remarks>
        /// <para>
        /// Each <see cref="IOutputAdapter"/> implementation reports its current queue size of unprocessed
        /// measurements so that if queue size reaches an unhealthy threshold, host can take evasive action.
        /// </para>
        /// <para>
        /// <see cref="EventArgs{T}.Argument"/> is total number of unprocessed measurements.
        /// </para>
        /// </remarks>
        public event EventHandler<EventArgs<int>> UnprocessedMeasurements;

        /// <summary>
        /// Indicates to the host that processing for one of the input adapters has completed.
        /// </summary>
        /// <remarks>
        /// This event is expected to only be raised when an input adapter has been designed to process
        /// a finite amount of data, e.g., reading a historical range of data during temporal procesing.
        /// </remarks>
        public event EventHandler ProcessingComplete;

        /// <summary>
        /// Event is raised when this <see cref="LegacyAdapterCollectionBase{T}"/> is disposed or an <see cref="ILegacyAdapter"/> in the collection is disposed.
        /// </summary>
        public event EventHandler Disposed;

        // Fields
        private Guid m_nodeID;
        private IList<NetworkCommunicator> m_networkCommunicators;
        private InputAdapterCollection m_inputAdapters;
        private ActionAdapterCollection m_actionAdapters;
        private OutputAdapterCollection m_outputAdapters;
        private DataSet m_dataSource;
        private ConcurrentDictionary<object, string> m_derivedNameCache;
        private MeasurementKey[] m_inputMeasurementKeysRestriction;
        private int m_measurementWarningThreshold;
        private int m_measurementDumpingThreshold;
        private int m_defaultSampleSizeWarningThreshold;
        private string m_name;
        private bool m_disposed;
        private ConfigGraph m_graph;
        private IMessageBus m_messageBus;
        private Dictionary<IDataProcessor, Thread> m_containers;
        private NotificationDispatcher m_dispatcher;

        // Constants
        private const string StrLogSource = "IaonSession";
        #endregion

        #region [ Constructors ]

        /// <summary>
        /// Creates a new <see cref="IaonSession"/>.
        /// </summary>
        public IaonSession()
        {
            ConfigurationFile configFile = ConfigurationFile.Current;

            // Initialize system settings
            CategorizedSettingsElementCollection systemSettings = configFile.Settings["systemSettings"];

            systemSettings.Add("NodeID", Guid.NewGuid().ToString(), "Unique Node ID");

            m_nodeID = systemSettings["NodeID"].ValueAs<Guid>();

            // Initialize threshold settings
            CategorizedSettingsElementCollection thresholdSettings = configFile.Settings["thresholdSettings"];

            thresholdSettings.Add("MeasurementWarningThreshold", "100000", "Number of unarchived measurements allowed in any output adapter queue before displaying a warning message");
            thresholdSettings.Add("MeasurementDumpingThreshold", "500000", "Number of unarchived measurements allowed in any output adapter queue before taking evasive action and dumping data");
            thresholdSettings.Add("DefaultSampleSizeWarningThreshold", "10", "Default number of unpublished samples (in seconds) allowed in any action adapter queue before displaying a warning message");

            m_measurementWarningThreshold = thresholdSettings["MeasurementWarningThreshold"].ValueAsInt32();
            m_measurementDumpingThreshold = thresholdSettings["MeasurementDumpingThreshold"].ValueAsInt32();
            m_defaultSampleSizeWarningThreshold = thresholdSettings["DefaultSampleSizeWarningThreshold"].ValueAsInt32();

            // Create a cache for derived adapter names
            m_derivedNameCache = new ConcurrentDictionary<object, string>();

            // Create network communicators collection
            m_networkCommunicators = new List<NetworkCommunicator>();

            // Create input adapters collection
            m_inputAdapters = new InputAdapterCollection();
            m_inputAdapters.ProcessingComplete += ProcessingCompleteHandler;

            // Create action adapters collection
            m_actionAdapters = new ActionAdapterCollection();
            m_actionAdapters.UnpublishedSamples += UnpublishedSamplesHandler;

            // Create output adapters collection
            m_outputAdapters = new OutputAdapterCollection();
            m_outputAdapters.UnprocessedMeasurements += UnprocessedMeasurementsHandler;

            m_graph = new ConfigGraph();
            // TODO: fix this to be an option. Note -- must be power of 2
            m_messageBus = new ChannelMessageBus(2 << 10);
            NotificationDispatcher = new NotificationDispatcher();
        }

        /// <summary>
        /// Releases the unmanaged resources before the <see cref="IaonSession"/> object is reclaimed by <see cref="GC"/>.
        /// </summary>
        ~IaonSession()
        {
            Dispose(false);
        }

        #endregion

        #region [ Properties ]

        public virtual IList<NetworkCommunicator> NetworkCommunicators
        {
            get
            {
                return m_networkCommunicators;
            }
        }

        /// <summary>
        /// Gets the input adapter collection for this <see cref="IaonSession"/>.
        /// </summary>
        public virtual InputAdapterCollection InputAdapters
        {
            get
            {
                return m_inputAdapters;
            }
        }

        /// <summary>
        /// Gets the action adapter collection for this <see cref="IaonSession"/>.
        /// </summary>
        public virtual ActionAdapterCollection ActionAdapters
        {
            get
            {
                return m_actionAdapters;
            }
        }

        /// <summary>
        /// Gets the output adapter collection for this <see cref="IaonSession"/>.
        /// </summary>
        public virtual OutputAdapterCollection OutputAdapters
        {
            get
            {
                return m_outputAdapters;
            }
        }

        /// <summary>
        /// Gets or sets a routing table restriction for a collection of input measurement keys.
        /// </summary>
        public virtual MeasurementKey[] InputMeasurementKeysRestriction
        {
            get
            {
                return m_inputMeasurementKeysRestriction;
            }
            set
            {
                m_inputMeasurementKeysRestriction = value;
            }
        }

        // TODO: Signal
        /// <summary>
        /// Gets or sets the configuration <see cref="DataSet"/> for this <see cref="IaonSession"/>.
        /// </summary>
        public virtual DataSet DataSource
        {
            get
            {
                return m_dataSource;
            }
            set
            {
                m_dataSource = value;
            }
        }

        /// <summary>
        /// Gets or sets the <see cref="Guid"/> node ID  for this <see cref="IaonSession"/>.
        /// </summary>
        public virtual Guid NodeID
        {
            get
            {
                return m_nodeID;
            }
            set
            {
                m_nodeID = value;
            }
        }

        protected NotificationDispatcher NotificationDispatcher
        {
            get;
            set;
        }

        /// <summary>
        /// Gets name assigned to this <see cref="IaonSession"/>, if any.
        /// </summary>
        public virtual string Name
        {
            get
            {
                return m_name.ToNonNullString();
            }
            set
            {
                m_name = value;
            }
        }

        /// <summary>
        /// Gets the combined status of the adapters in this <see cref="IaonSession"/>.
        /// </summary>
        public virtual string Status
        {
            get
            {
                StringBuilder status = new StringBuilder();

                status.AppendLine();
                status.AppendLine(">> Input Adapters:");
                status.AppendLine();

                if (m_inputAdapters != null)
                    status.AppendLine(m_inputAdapters.Status);

                status.AppendLine();
                status.AppendLine(">> Action Adapters:");
                status.AppendLine();

                if (m_actionAdapters != null)
                    status.AppendLine(m_actionAdapters.Status);

                status.AppendLine();
                status.AppendLine(">> Output Adapters:");
                status.AppendLine();

                if (m_outputAdapters != null)
                    status.AppendLine(m_outputAdapters.Status);

                return status.ToString();
            }
        }

        #endregion

        #region [ Methods ]

        public virtual bool TryGetAdapterByID(uint id, out IAdapter adapter)
        {
            ILegacyAdapterCollection[] collections = { m_inputAdapters, m_actionAdapters, m_outputAdapters };

            foreach (ILegacyAdapterCollection collection in collections)
            {
                if (collection.TryGetAdapterByID(id, out adapter))
                    return true;
            }

            adapter = null;
            return false;
        }

        public virtual bool TryGetAdapterByName(string name, out IAdapter adapter)
        {
            ILegacyAdapterCollection[] collections = { m_inputAdapters, m_actionAdapters, m_outputAdapters };

            foreach (ILegacyAdapterCollection collection in collections)
            {
                if (collection.TryGetAdapterByName(name, out adapter))
                    return true;
            }

            adapter = null;
            return false;
        }

        /// <summary>
        /// Attempts to create an <see cref="IAdapterInfo"/> from the specified <see cref="DataRow"/>.
        /// </summary>
        /// <param name="adapterRow"><see cref="DataRow"/> containing item information to initialize.</param>
        /// <param name="info">Initialized adapter if successful; otherwise null.</param>
        /// <returns><c>true</c> if item was successfully initialized; otherwise <c>false</c>.</returns>
        /// <remarks>
        /// See <see cref="DataSource"/> property for expected <see cref="DataRow"/> column names.
        /// </remarks>
        /// <exception cref="NullReferenceException"><paramref name="adapterRow"/> is null.</exception>
        protected virtual bool TryCreateInfo(DataRow adapterRow, out IConfigInfo info)
        {
            if (adapterRow == null)
                throw new NullReferenceException(string.Format("Cannot initialize from null adpater DataRow"));
            info = null;
            Assembly assembly;
            Type adapterType;
            string name = "", assemblyName = "", typeName = "", connectionString, setting;
            uint id;

            try
            {
                name = adapterRow["AdapterName"].ToNonNullString("[ILegacyAdapter]");
                assemblyName = FilePath.GetAbsolutePath(adapterRow["AssemblyName"].ToNonNullString());
                typeName = adapterRow["TypeName"].ToNonNullString();
                connectionString = adapterRow["ConnectionString"].ToNonNullString();
                id = uint.Parse(adapterRow["ID"].ToNonNullString("0"));

                if (string.IsNullOrWhiteSpace(typeName))
                    throw new InvalidOperationException("No adapter type was defined");

                if (!File.Exists(assemblyName))
                    throw new InvalidOperationException("Specified adapter assembly does not exist");

                assembly = Assembly.LoadFrom(assemblyName);
                adapterType = assembly.GetType(typeName);
                foreach (Type nestedTypes in adapterType.GetNestedTypes())
                {
                    if (typeof(IConfigInfo).IsAssignableFrom(nestedTypes))
                    {
                        info = (IConfigInfo)Activator.CreateInstance(nestedTypes);
                        //HACK what if there are 2 info types, other than super duper brokenness
                        break;
                    }
                }
                if (info == null)
                {
                    throw new InitializationException(String.Format("No info type Found for {0}:{1}", assemblyName, name));
                }
                LegacyAdapterBase.AdapterInfoBase legacyInfo = info as LegacyAdapterBase.AdapterInfoBase;


                info.Update(GsfHandler.ParseConnectionString(connectionString));

                // Assign critical adapter properties
                info.Name = name;
                info.ID = id;
                info.RuntimeType = adapterType;

                // Note - this is a future step, SetConnectionString does this now, but we want what's below
                //adapter.Settings = GSFHandler.ParseConnectionString(connectionString);
                info.ExtractConfiguration(DataSource);
                IAdapterInfo ai = info as IAdapterInfo;
                if ((object)ai != null)
                {
                    // Assign adapter initialization timeout   
                    if (ai.Settings.TryGetValue("initializationTimeout", out setting))
                        ai.InitializationTimeout = int.Parse(setting);
                    else
                        // TODO: Fix placement
                        ai.InitializationTimeout = LegacyAdapterBase.DefaultInitializationTimeout;
                }
                info.Receiver = this.NotificationDispatcher.GetReceiver(info);
                return true;
            }
            catch (Exception ex)
            {
                // We report any errors encountered during type creation...
                LogDispatcher.E(StrLogSource, "ProcessException", "Failed to load adapter \"{0}\" [{1}] from \"{2}\": {3}", name, typeName, assemblyName, ex.Message);
                LogDispatcher.E(StrLogSource, "ProcessException", ex);
            }

            info = null;
            return false;
        }


        /// <summary>
        /// Releases all the resources used by the <see cref="IaonSession"/> object.
        /// </summary>
        public void Dispose()
        {
            Dispose(true);
            GC.SuppressFinalize(this);
        }

        /// <summary>
        /// Releases the unmanaged resources used by the <see cref="IaonSession"/> object and optionally releases the managed resources.
        /// </summary>
        /// <param name="disposing">true to release both managed and unmanaged resources; false to release only unmanaged resources.</param>
        protected virtual void Dispose(bool disposing)
        {
            if (!m_disposed)
            {
                try
                {
                    if (disposing)
                    {
                        DataSet dataSource = this.DataSource;

                        // Dispose input adapters collection
                        if (m_inputAdapters != null)
                        {
                            m_inputAdapters.ProcessingComplete -= ProcessingCompleteHandler;
                            m_inputAdapters.Dispose();
                        }
                        m_inputAdapters = null;

                        // Dispose action adapters collection
                        if (m_actionAdapters != null)
                        {
                            m_actionAdapters.UnpublishedSamples -= UnpublishedSamplesHandler;
                            m_actionAdapters.Dispose();
                        }
                        m_actionAdapters = null;

                        // Dispose output adapters collection
                        if (m_outputAdapters != null)
                        {
                            m_outputAdapters.UnprocessedMeasurements -= UnprocessedMeasurementsHandler;
                            m_outputAdapters.Dispose();
                        }
                        m_outputAdapters = null;

                        if ((object)dataSource != null)
                            dataSource.Dispose();
                    }
                }
                finally
                {
                    m_disposed = true;  // Prevent duplicate dispose.

                    if (Disposed != null)
                        Disposed(this, EventArgs.Empty);
                }
            }
        }

        /// <summary>
        /// Initialize and start adapters.
        /// </summary>
        public virtual void Initialize()
        {
            IConfigInfo item;
            m_dispatcher = new NotificationDispatcher();

            List<IConfigInfo> allInfos = new List<IConfigInfo>();
            //note - this is a temporary solution, as things shake out we should look to delete this.
            // KEY: adapterCollection, VALUE: List of config infos that go in that collection.
            Dictionary<IList, List<IConfigInfo>> infosLookup = new Dictionary<IList, List<IConfigInfo>>();
            // make all our infos from the relevant data source bits track infos by type because 
            // adapter collections need the instnatiated bits
            foreach (
                ILegacyAdapterCollection targetCollection in
                    new ILegacyAdapterCollection[] { m_actionAdapters, m_inputAdapters, m_outputAdapters })
            {
                if (DataSource.Tables.Contains(targetCollection.DataMember))
                {
                    List<IConfigInfo> infoCollection = new List<IConfigInfo>();
                    infosLookup.Add((IList)targetCollection, infoCollection);
                    foreach (DataRow adapterRow in DataSource.Tables[targetCollection.DataMember].Rows)
                    {
                        if (TryCreateInfo(adapterRow, out item))
                        {
                            allInfos.Add(item);
                            infoCollection.Add(item);
                        }
                    }
                }
                else
                    throw new InvalidOperationException(
                        string.Format(
                            "Data set member \"{0}\" was not found in data source, check ConfigurationEntity. Failed to initialize {1}.",
                            targetCollection.DataMember, targetCollection.Name));

            }
            foreach (DataRow infoRow in DataSource.Tables["NetworkCommunicators"].Rows)
            {
                if (TryCreateInfo(infoRow, out item))
                {
                    allInfos.Add(item);
                }

            }
            // Make infos for nework communicators
            // factor into a command
            m_graph.Build(allInfos);
            m_graph.Collapse(1);
            m_containers = new Dictionary<IDataProcessor, Thread>();
            // Collapsing of the graph makes ContainerInfos holding the Relevant AdapterInfo, if its a container,
            // otherwise if it's a NetworkCommunicator, makes it an info of that type.
            //TODO: node should be updated for network communicators to- IDataProcessorInfo instead.
            foreach (IDataProcessorInfo containerInfo in m_graph.Nodes)
            {
                if ((object)containerInfo.Receiver == null)
                {
                    containerInfo.Receiver = NotificationDispatcher.GetReceiver(containerInfo);
                }
                // HACK: Still relies on the AdapterInfo and Adapter having references to each other
                List<string> names = new List<string>();
                // If no member infos (e.g. network communicator) this doesn't happen.
                // HACK: figure bettter way do deal with members...
                if (containerInfo is IAdapterContainerInfo)
                {
                    foreach (IAdapterInfo node in ((IAdapterContainerInfo)containerInfo).MemberInfos)
                    {
                        IAdapter adapterInstance =
                            (IAdapter)Activator.CreateInstance(node.RuntimeType, node);
                        ILegacyAdapter legacyAdapter = adapterInstance as ILegacyAdapter;

                        if ((object)legacyAdapter != null)
                        {
                            legacyAdapter.Name = legacyAdapter.Info.Name;
                            legacyAdapter.ID = legacyAdapter.Info.ID;
                            legacyAdapter.UseLegacyCommunication = false;
                        }
                        names.Add(node.Name);
                    }
                }

                // HACK: make a better container/mbp name. (of course that name isn't actually used, maybe just
                // kill that or figure out how it's used for logging.
                // make data processor
                IDataProcessor thisprocessor =
                    (IDataProcessor)Activator.CreateInstance(containerInfo.RuntimeType, m_messageBus.GetProxy(string.Format("{0}", string.Join(",", names))), containerInfo);

                m_containers.Add(thisprocessor, null);
                thisprocessor.Register();
            }

            // now add each instance to the relevant collection
            foreach (IList collection in infosLookup.Keys)
            {
                foreach (IAdapterInfo info in infosLookup[collection])
                {
                    // This also calls Adapter.Initialize() - revisit
                    collection.Add(info.MyInstance);
                }
            }

            // Spawn a thread to call doit() on all adapter containers so that
            // adapters can process measurements that are placed on the message bus.

            // TODO: do we want to have a circular AdapterContainerInfo <-> AdapterContianer reference,
            // like we do with Adapters and AdapterInfos? if so, we don't strictly need this
            //  but... circular references are a code smell revist with big picture shape in mind
            LogDispatcher.D("IaonSession", "Initialize", "Starting data processing threads...");

            foreach (IDataProcessor container in m_containers.Keys.ToArray())
            {
                NetworkCommunicator networkCommunicator = container as NetworkCommunicator;

                if ((object)networkCommunicator != null)
                    m_networkCommunicators.Add(networkCommunicator);
                
                // For closure.
                Thread doitThread = new Thread(container.Start);
                m_containers[container] = doitThread;
                doitThread.IsBackground = true;
                doitThread.Start();
            }
        }

        /// <summary>
        /// Gets derived name of specified object.
        /// </summary>
        /// <param name="sender">Sending object from which to derive name.</param>
        /// <returns>Derived name of specified object.</returns>
        public virtual string GetDerivedName(object sender)
        {
            return m_derivedNameCache.GetOrAdd(sender, key =>
            {
                string name = null;
                IProvideStatus statusProvider = key as IProvideStatus;

                if (statusProvider != null)
                    name = statusProvider.Name;
                else if (key != null && key is string)
                    name = (string)key;

                if (string.IsNullOrWhiteSpace(name))
                    name = key.GetType().Name;

                if (!string.IsNullOrWhiteSpace(m_name))
                    name += "#" + m_name;

                return name;
            });
        }

        /// <summary>
        /// Raises <see cref="InputMeasurementKeysUpdated"/> event.
        /// </summary>
        /// <param name="sender">Object source raising the event.</param>
        protected virtual void OnInputMeasurementKeysUpdated(object sender)
        {
            if (InputMeasurementKeysUpdated != null)
                InputMeasurementKeysUpdated(sender, EventArgs.Empty);
        }

        /// <summary>
        /// Raises <see cref="OutputMeasurementsUpdated"/> event.
        /// </summary>
        /// <param name="sender">Object source raising the event.</param>
        protected virtual void OnOutputMeasurementsUpdated(object sender)
        {
            if (OutputMeasurementsUpdated != null)
                OutputMeasurementsUpdated(sender, EventArgs.Empty);
        }

        /// <summary>
        /// Raises the <see cref="UnpublishedSamples"/> event.
        /// </summary>
        /// <param name="sender">Object source raising the event.</param>
        /// <param name="seconds">Total number of unpublished seconds of data.</param>
        protected virtual void OnUnpublishedSamples(object sender, int seconds)
        {
            if (UnpublishedSamples != null)
                UnpublishedSamples(sender, new EventArgs<int>(seconds));
        }

        /// <summary>
        /// Raises the <see cref="UnprocessedMeasurements"/> event.
        /// </summary>
        /// <param name="sender">Object source raising the event.</param>
        /// <param name="unprocessedMeasurements">Total measurements in the queue that have not been processed.</param>
        protected virtual void OnUnprocessedMeasurements(object sender, int unprocessedMeasurements)
        {
            if (UnprocessedMeasurements != null)
                UnprocessedMeasurements(sender, new EventArgs<int>(unprocessedMeasurements));
        }

        /// <summary>
        /// Raises the <see cref="ProcessingComplete"/> event.
        /// </summary>
        /// <param name="sender">Object source raising the event.</param>
        /// <param name="e"><see cref="EventArgs"/>, if any.</param>
        protected virtual void OnProcessingComplete(object sender, EventArgs e = null)
        {
            if (ProcessingComplete != null)
                ProcessingComplete(sender, e ?? EventArgs.Empty);
        }

        /// <summary>
        /// Raises the <see cref="Disposed"/> event.
        /// </summary>
        /// <param name="sender">Object source raising the event.</param>
        protected virtual void OnDisposed(object sender)
        {
            if (Disposed != null)
                Disposed(sender, EventArgs.Empty);
        }

        /// <summary>
        /// Event handler for reporting status messages.
        /// </summary>
        /// <param name="sender">Event source of the status message.</param>
        /// <param name="e">Event arguments containing the status message to report.</param>
        public virtual void StatusMessageHandler(object sender, EventArgs<string> e)
        {
            LogDispatcher.I(StrLogSource, "StatusMessage", "[{0}] {1}", GetDerivedName(sender), e.Argument);
        }

        /// <summary>
        /// Event handler for processing reported exceptions.
        /// </summary>
        /// <param name="sender">Event source of the exception.</param>
        /// <param name="e">Event arguments containing the exception to report.</param>
        public virtual void ProcessExceptionHandler(object sender, EventArgs<Exception> e)
        {
            LogDispatcher.E(StrLogSource, "StatusMessage", "[{0}] {1}", GetDerivedName(sender), e.Argument.Message);
            LogDispatcher.E(StrLogSource, "StatusMessage", e.Argument);
            // TODO: Also do ProcessException
        }

        /// <summary>
        /// Event handler for monitoring unpublished samples.
        /// </summary>
        /// <param name="sender">Event source reference to adapter, typically an action adapter, that is reporting the number of unpublished data samples.</param>
        /// <param name="e">Event arguments containing number of samples, in seconds of data, of unpublished data in the source adapter.</param>
        /// <remarks>
        /// Time-series framework uses this handler to monitor the number of unpublished samples, in seconds of data, in action adapters.<br/>
        /// This method is typically called once every five seconds.
        /// </remarks>
        public virtual void UnpublishedSamplesHandler(object sender, EventArgs<int> e)
        {
            int secondsOfData = e.Argument;
            int threshold = m_defaultSampleSizeWarningThreshold;
            int processingInterval = -1;
            ConcentratorBase concentrator = sender as ConcentratorBase;
            ILegacyAdapter adapter = sender as ILegacyAdapter;

            // Most action adapters will be based on a concentrator, if so we monitor the unpublished sample queue size compared to the defined
            // lag time - if the queue size is over twice the lag size, the action adapter could be falling behind
            if ((object)concentrator != null)
                threshold = (int)(2 * Math.Ceiling(concentrator.LagTime));

            // Get processing interval for adapter
            if ((object)adapter != null)
                processingInterval = adapter.ProcessingInterval;

            // Allow much more time before warning for when a fast processing interval has been defined
            if (processingInterval > -1 && processingInterval < 100)
                threshold *= 4;

            if (secondsOfData > threshold)
                LogDispatcher.I(StrLogSource, "StatusMessage", "[{0}] There are {1} seconds of unpublished data in the action adapter concentration queue.", GetDerivedName(sender), secondsOfData);

            // Bubble message up to any event subscribers
            OnUnpublishedSamples(sender, e.Argument);
        }

        /// <summary>
        /// Event handler for monitoring unprocessed measurements.
        /// </summary>
        /// <param name="sender">Event source reference to adapter, typically an output adapter, that is reporting the number of unprocessed measurements.</param>
        /// <param name="e">Event arguments containing number of queued (i.e., unprocessed) measurements in the source adapter.</param>
        /// <remarks>
        /// Time-series framework uses this handler to monitor the number of unprocessed measurements in output adapters.<br/>
        /// This method is typically called once every five seconds.
        /// </remarks>
        public virtual void UnprocessedMeasurementsHandler(object sender, EventArgs<int> e)
        {
            int unprocessedMeasurements = e.Argument;

            if (unprocessedMeasurements > m_measurementDumpingThreshold)
            {
                ILegacyOutputAdapter outputAdapter = sender as ILegacyOutputAdapter;

                if (outputAdapter != null)
                {
                    // If an output adapter queue size exceeds the defined measurement dumping threshold,
                    // then the queue will be truncated before system runs out of memory
                    outputAdapter.RemoveMeasurements(m_measurementDumpingThreshold);
                    LogDispatcher.E(StrLogSource, "StatusMessage",
                                    "[{0}] System exercised evasive action to conserve memory and dumped {1} unprocessed measurements from the output queue :(",
                                    outputAdapter.Name, m_measurementDumpingThreshold);
                    LogDispatcher.I(StrLogSource, "StatusMessage", "[{0}] NOTICE: Please adjust measurement threshold settings and/or increase amount of available system memory.", outputAdapter.Name);
                }
                else
                {
                    // It is only expected that output adapters will be mapped to this handler, but in case
                    // another adapter type uses this handler we will still display a message
                    LogDispatcher.E(StrLogSource, "StatusMessage", "[{0}] CRITICAL: There are {1} unprocessed measurements in the adapter queue - but sender \"{2}\" is not an IOutputAdapter, so no evasive action can be exercised.", GetDerivedName(sender), unprocessedMeasurements, sender.GetType().Name);
                }
            }
            else if (unprocessedMeasurements > m_measurementWarningThreshold)
            {
                if (unprocessedMeasurements >= m_measurementDumpingThreshold - m_measurementWarningThreshold)
                    LogDispatcher.E(StrLogSource, "StatusMessage",
                                    "[{0}] CRITICAL: There are {1} unprocessed measurements in the output queue.",
                                    UpdateType.Warning, GetDerivedName(sender), unprocessedMeasurements);
                else
                    LogDispatcher.I(StrLogSource, "StatusMessage",
                                    "[{0}] There are {1} unprocessed measurements in the output queue.",
                                    GetDerivedName(sender), unprocessedMeasurements);
            }

            // Bubble message up to any event subscribers
            OnUnprocessedMeasurements(sender, e.Argument);
        }

        /// <summary>
        /// Event handler for processing complete notifications from input adapters.
        /// </summary>
        /// <param name="sender">Event source reference to input adapter that is reporting processing completion.</param>
        /// <param name="e">Event arguments for event, if any; otherwise <see cref="EventArgs.Empty"/>.</param>
        public virtual void ProcessingCompleteHandler(object sender, EventArgs e)
        {
            LogDispatcher.I(StrLogSource, "StatusMessage", "[{0}] Processing completed.", GetDerivedName(sender));

            // Bubble message up to any event subscribers
            OnProcessingComplete(sender, e);
        }

        /// <summary>
        /// Event handler for disposed events from all adapters.
        /// </summary>
        /// <param name="sender">Sending object.</param>
        /// <param name="e">Event arguments, if any.</param>
        public virtual void DisposedHandler(object sender, EventArgs e)
        {
            LogDispatcher.I(StrLogSource, "StatusMessage", "[{0}] Disposed.", GetDerivedName(sender));

            // Bubble message up to any event subscribers
            OnDisposed(sender);
        }

        // Bubble routing table exceptions out through Iaon session
        private void m_routingTables_ProcessException(object sender, EventArgs<Exception> e)
        {
            ProcessExceptionHandler(sender, e);
        }

        #endregion

        #region [ Static ]

        // Static Methods

        /// <summary>
        /// Extracts a configuration that supports temporal processing from an existing real-time configuration.
        /// </summary>
        /// <param name="realtimeConfiguration">Real-time <see cref="DataSet"/> configuration.</param>
        /// <returns>A new <see cref="DataSet"/> configuration for adapters that support temporal processing.</returns>
        public static DataSet ExtractTemporalConfiguration(DataSet realtimeConfiguration)
        {
            // Duplicate current run-time session configuration that has temporal support
            DataSet temporalConfiguration = new DataSet("IaonTemporal");
            DataTable temporalSupport = realtimeConfiguration.Tables["TemporalSupport"];
            string tableName;

            foreach (DataTable table in realtimeConfiguration.Tables)
            {
                tableName = table.TableName;

                switch (tableName.ToLower())
                {
                    case "inputadapters":
                    case "actionadapters":
                    case "outputadapters":
                        // For Iaon adapter tables, we only copy in adapters that support temporal processing
                        temporalConfiguration.Tables.Add(table.Clone());

                        // Check for adapters with temporal support in this adapter collection
                        DataRow[] temporalAdapters = temporalSupport.Select(string.Format("Source = '{0}'", tableName));

                        // If any adapters support temporal processing, add them to the temporal configuration
                        if (temporalAdapters.Length > 0)
                        {
                            DataTable realtimeTable = realtimeConfiguration.Tables[tableName];
                            DataTable temporalTable = temporalConfiguration.Tables[tableName];

                            foreach (DataRow row in realtimeTable.Select(string.Format("ID IN ({0})", temporalAdapters.Select(row => row["ID"].ToString()).ToDelimitedString(','))))
                            {
                                DataRow newRow = temporalTable.NewRow();

                                for (int x = 0; x < realtimeTable.Columns.Count; x++)
                                {
                                    newRow[x] = row[x];
                                }

                                temporalConfiguration.Tables[tableName].Rows.Add(newRow);
                            }
                        }

                        break;
                    default:
                        // For all other tables we add configuration information as-is
                        temporalConfiguration.Tables.Add(table.Copy());
                        break;
                }
            }

            return temporalConfiguration;
        }

        #endregion
    }
}